package cn.wolfcode.queue.rabbitmq.rabbitmqspringboot.producer;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * @author: wangzhe
 * @create: 2020/9/23 11:15 上午
 * @Description 队列消息发送基类
 */
@Slf4j
@Component
public class Producer {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public boolean sendNormalQueue(Object obj, String exchange, String routerKey) {
        return sendNormalQueue(obj, exchange, routerKey, null, null);
    }

    public boolean sendNormalQueue(Object obj, String exchange, String routerKey, RabbitTemplate.ConfirmCallback confirmCallback) {
        return sendNormalQueue(obj, exchange, routerKey, confirmCallback, null);
    }

    /**
     * 发送常规队列消息
     *
     * @param obj
     * @param confirmCallback
     * @param correlationData
     * @param confirmCallback
     * @param correlationData
     * @return
     */
    public boolean sendNormalQueue(Object obj, String exchange, String routerKey, RabbitTemplate.ConfirmCallback confirmCallback, CorrelationData correlationData) {
        try {
            rabbitTemplate.setConfirmCallback(confirmCallback);
            rabbitTemplate.convertAndSend(exchange, routerKey, obj, message -> {
                return message;
            }, correlationData);
        } catch (Exception e) {
            log.error("发送常规队列消息失败，MSG:" + JSON.toJSONString(obj), e);
            return false;
        }
        return true;
    }


    public boolean sendNormalTtlQueue(Object messageContent, final long delayTimes, String exchange, String routerKey) {
        return sendNormalTtlQueue(messageContent, delayTimes, exchange, routerKey, null, null);
    }

    public boolean sendNormalTtlQueue(Object messageContent, final long delayTimes, String exchange, String routerKey, RabbitTemplate.ConfirmCallback confirmCallback) {
        return sendNormalTtlQueue(messageContent, delayTimes, exchange, routerKey, confirmCallback, null);
    }


    /**
     * 灵活队列发送延迟消息
     *
     * @param messageContent  消息内容
     * @param exchange        队列交换
     * @param routerKey       队列交换绑定的路由键
     * @param delayTimes      延迟时长，单位：毫秒
     * @param confirmCallback
     * @param correlationData
     */
    public boolean sendNormalTtlQueue(Object messageContent, final long delayTimes, String exchange, String routerKey, RabbitTemplate.ConfirmCallback confirmCallback, CorrelationData correlationData) {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        log.info("延迟消息发送时间:" + sdf.format(new Date()) + "，{}毫秒写入消息队列：{}，消息内容：{}", delayTimes, routerKey,
                JSON.toJSONString(messageContent));
        if (!StringUtils.isEmpty(exchange) && messageContent != null) {
            try {
                // 执行发送消息到指定队列
                rabbitTemplate.convertAndSend(exchange, routerKey, messageContent, message -> {
                    // 设置延迟毫秒值 TTL，Time-To-Live Extensions（过期时间）： RabbitMQ 允许你对 message 和 queue 设置 TTL 值。
                    // expiration 字段以毫秒为单位表示 TTL的值
                    message.getMessageProperties().setExpiration(String.valueOf(delayTimes));
                    return message;
                }, correlationData);
            } catch (Exception e) {
                log.error("发送延迟消息失败，MSG:" + messageContent, e);
                return false;
            }
            return true;
        } else {
            log.error("未找到队列消息：{}，所属的交换机或者消息为null", exchange);
            return false;
        }
    }


}